package com.example.reactor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedDeque;

/**
 * 当client连接服务器发起操作后，服务器必须全部处理完
 * 整个交互才算结束，如果没有全部处理完成，select方法就会被一只调用
 * 1.数据没有处理完整前select方法就会被多次调用
 * 2.在某些特殊的操作下，服务端无法处理，那么select方法就会被频繁调用，
 * 调用selectKey.cancel()等同于处理client功能，避免再次调用select方法
 */
public class Worker implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(Worker.class);
    private Selector selector;
    private Thread thread;
    private String name;
    private volatile boolean isCreated;
    private final ConcurrentLinkedDeque<Runnable> deque = new ConcurrentLinkedDeque<>();


    public Worker(String name) {
        this.name = name;
    }

    public void register(SocketChannel socketChannel) throws IOException {
        if (!isCreated) {
            selector = Selector.open();
            thread = new Thread(this, name);
            thread.start();  //  初始化的并且启动线程
            isCreated = true;
        }
        socketChannel.configureBlocking(false);
        // 必须要将register和select放在一个线程里面
        // 将任务放到队列里面
        deque.add(() -> {
            try {
                socketChannel.register(selector, SelectionKey.OP_READ);
            } catch (ClosedChannelException e) {
                log.error("error", e);
            }
        });
        selector.wakeup();

    }

    @Override
    public void run() {
        try {
            while (true) {
                selector.select();
                // 从队列里面拉取任务进行注册
                Runnable runnable = deque.poll();
                if (runnable != null) {
                    runnable.run();
                }
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();
                    if (selectionKey.isReadable()) {
                        ByteBuffer buffer = ByteBuffer.allocate(20);
                        SocketChannel channel = (SocketChannel) selectionKey.channel();
                        int read = channel.read(buffer);
                        if (read == -1) {
                            selectionKey.cancel();
                        } else {
                            buffer.flip();
                            log.info("msg is:{}", Charset.defaultCharset().decode(buffer));
                            buffer.clear();
                        }
                    }
                }
            }

        } catch (IOException e) {
            log.error("error:", e);
        }

    }
}
